package xu.scala

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}


object SparkKafka {
  def main(args: Array[String]): Unit = {
    //spark streaming 实现 kafka 的消费者
    //1) 构建sparkconf 本地运行 ，运行应用程序名称
    //2) 构建sparkstreaming --->> streamingContext， 加载配置
    //3) kafka 配置 broker ， key, value， group id 消费模式
    //4) spark 链接 kafka 订阅 ， topic， streamingcontext
    //5) 循环的形式 打印/ 处理
    //6) 开启ssc， 监控 kafka 数据

    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkKafka")

    //StreamingContext 需要导入依赖
    //spark streaming 可以进行流式处理 ，微批次处理 ,间隔2秒
    val ssc = new StreamingContext(conf, Seconds(2))

    //spark info 输出红色 info信息  -->> error才是报错
    ssc.sparkContext.setLogLevel("error")

    //kafka 配置
    val kfkaParram = Map[String, Object](
      "bootstrap.servers" -> "192.168.61.128:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "15homework",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //sparkstreaming 链接kafka
    val topicName = Array("15homework")
    val streamRdd = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topicName,kfkaParram)
    )

    //返回 kafka 一段有间隔的 streamRdd
    streamRdd.foreachRDD(
      x => {
        if(! x.isEmpty()){  //判断是否为空 !相反
          val line = x.map(_.value())
          line.foreach(println)
        }
      }
    )

    ssc.start()
    ssc.awaitTermination()

  }
}
